package cloud.pangui.core.worker;

import cloud.pangui.core.utils.ClassUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Auther hepangui
 * @Date 2020/11/7 0007
 */
public class WorkerManager {


	private static Logger logger = LoggerFactory.getLogger(WorkerManager.class);
	/**
	 * 最大工人数
	 */
	protected int maxSize = Runtime.getRuntime().availableProcessors() * 2;
	/**
	 * 保留数量
	 */
	protected int coreSize = 0;
	/**
	 * 最大空闲时间
	 */
	protected long maxIdle = 60 * 1000L;
	/**
	 * 最大消息数
	 */
	protected int maxMessage = Integer.MAX_VALUE;
	/**
	 * 传递到worker的message
	 */
	protected SynchronousQueue syncWorkMessage;

	/**
	 * 等待工作的message
	 */
	protected volatile LinkedBlockingQueue messages;

	/**
	 * 工人类
	 */
	protected Class<? extends Worker> workerClass;

	protected ConcurrentLinkedQueue<Worker> workers = new ConcurrentLinkedQueue<>();

	/**
	 * workerSize
	 */
	protected AtomicInteger workerCount = new AtomicInteger(0);

	/**
	 * 构造函数
	 */
	public WorkerManager() {
		syncWorkMessage = new SynchronousQueue();
		messages = new LinkedBlockingQueue();
	}

	/**
	 * 工人类
	 */
	public WorkerManager(Class workerClass) {
		this();
		this.workerClass = workerClass;
	}


	/**
	 * 添加消息，如果当前忙，不添加消息直接返回false
	 *
	 * @param message
	 */
	public boolean addMessage0(Object message) {
		//offer成功，标识当前有可用的线程
		boolean offerSuccess = this.syncWorkMessage.offer(message);
		if (!offerSuccess) {
			if (workerCount.get() < maxSize) {
				synchronized (workerCount) {
					if (workerCount.get() < maxSize) {
						if (workerCount.get() < coreSize) {
							getWorker(true, message);
						} else if (workerCount.get() < maxSize) {
							getWorker(false, message);
						}
					} else {
						return false;
					}
				}
			} else {
				return false;
			}
		}
		return true;
	}

	/**
	 * 添加消息，如果当前忙，则添加到队列中等待执行
	 * 如果队列已满，抛出异常
	 *
	 * @param message
	 */
	public void addMessage(Object message) {
		if (!addMessage0(message)) {
			messages.add(message);
		}
	}

	/**
	 * 获取工人
	 */
	public Worker getWorker(boolean keepAlive, Object firstMessage) {
		Worker worker = createWorker();
		worker.keepAlive = keepAlive;
		worker.firstMessage = firstMessage;
		worker.workerManager = this;
		this.workerCount.incrementAndGet();
		this.workers.add(worker);
		worker.start();
		return worker;
	}

	/**
	 * 线程闲时自动销毁
	 */
	public void removeDead(Worker worker) {
		workers.remove(worker);
	}

	/**
	 * 创建工人
	 */
	public Worker createWorker() {
		return ClassUtils.newInstance(this.workerClass);
	}


	/**
	 * 获取最大工人数
	 */
	public int getMaxSize() {
		return maxSize;
	}

	public void setMaxSize(int maxSize) {
		this.maxSize = maxSize;
	}

	/**
	 * 设置最大工人数
	 */
	public void resetMaxSize(int maxSize) {
		int maxWorkerOld = this.maxSize;
		this.maxSize = maxSize;
		if (maxWorkerOld > maxSize && this.workerCount.get() > maxSize) {

			int i = this.workerCount.get();
			for (Worker worker : this.workers) {
				if (!worker.working) {
					worker.getCurrentThread().interrupt();
					i--;
				}
			}
			if (i > maxSize) {
				int i1 = 0;
				for (Worker worker : this.workers) {
					if (i1 > i - maxSize) {
						break;
					}
					worker.getCurrentThread().interrupt();
					i1++;
				}
			}
		}
	}

	/**
	 * 获取最大消息数
	 */
	public int getMaxMessage() {
		return maxMessage;
	}

	/**
	 * 设置最大消息数
	 */
	public List<Object> setMaxMessage(int maxMessage) {

		//消息大小变化时，重新初始化队列,并将原有消息复制
		List<Object> list = new ArrayList<>();
		if (this.maxMessage != maxMessage) {
			this.maxMessage = maxMessage;
			BlockingQueue queue = this.messages;
			this.messages = new LinkedBlockingQueue(maxMessage);
			for (Object o : queue) {
				if (!this.messages.offer(o)) {
					list.add(o);
				}
			}
		}
		return list;
	}

	public int getCoreSize() {
		return coreSize;
	}

	public void setCoreSize(int coreSize) {
		this.coreSize = coreSize;
	}

	public long getMaxIdle() {
		return maxIdle;
	}

	public void setMaxIdle(long maxIdle) {
		this.maxIdle = maxIdle;
	}

	public int getMessageCount() {
		return messages.size();
	}

	public ConcurrentLinkedQueue<Worker> getWorkers() {
		return workers;
	}

	public LinkedBlockingQueue getMessages() {
		return messages;
	}

	public int getWorkerCount() {
		return workerCount.get();
	}


	/**
	 * 销毁
	 */
	public void destory() {
		this.messages.clear();
		for (Worker worker : this.workers) {
			worker.destory();
		}
	}

	/**
	 * 是否忙(有待处理队列,Worker正在处理)
	 */
	public boolean isBusy() {
		if (messages.size() > 0) {
			return true;
		}

		for (Worker worker : this.workers) {
			if (worker.isWorking()) {
				return true;
			}
		}
		return false;
	}

	public int getBusyWorkerCount() {
		int i = 0;
		for (Worker worker : this.workers) {
			if (worker.isWorking()) {
				i++;
			}
		}
		return i;
	}

	protected SynchronousQueue getSyncWorkMessage() {
		return syncWorkMessage;
	}

}
